package com.hudson.broker.client;


import com.hudson.broker.config.BrokerConfig;
import com.hudson.broker.monitor.ThreadPoolHelper;
import com.hudson.broker.handler.netty.ReconnectionHandler;
import com.hudson.broker.utils.NettyUtil;
import com.hudson.codec.MessageCodecFactory;
import com.hudson.codec.domain.RemotingCommand;
import com.hudson.codec.proto.domain.RemotingCommandProto;
import com.hudson.common.domain.meta.BrokerInfo;
import com.hudson.common.enums.MessageTypeEnum;
import com.hudson.core.json.JSONUtil;
import com.hudson.store.queue.QueueNumUtil;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.net.InetSocketAddress;
import java.nio.charset.StandardCharsets;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * &#064;Author:  渔小鲤
 * &#064;DateTime:  2025/7/12 20:18
 **/
public class NameSrvClient {
    private static final Logger logger = LoggerFactory.getLogger(NameSrvClient.class);
    private static final ConcurrentHashMap<String, AtomicInteger> RETRY_NUM_MAP = new ConcurrentHashMap<>();
    private static final ConcurrentHashMap<String, Channel> NAME_SRV_CHANNEL_MAP = new ConcurrentHashMap<>();
    private static final Bootstrap client = new Bootstrap().group(ThreadPoolHelper.NAME_SRV_WORK_EVENT_LOOP).channel(NioSocketChannel.class).option(ChannelOption.TCP_NODELAY, true).handler(new ChannelInitializer<NioSocketChannel>() {
        @Override
        protected void initChannel(NioSocketChannel sc) {
            sc.pipeline().addLast(new LengthFieldBasedFrameDecoder(1024 * 1024, 8, 4, 0, 0));
            sc.pipeline().addLast(MessageCodecFactory.getCodec(BrokerConfig.serializableMethod));
            sc.pipeline().addLast(new IdleStateHandler(0, 10, 0, TimeUnit.SECONDS)); // 写空闲4秒触发 WRITE_IDLE
            sc.pipeline().addLast(new ChannelInboundHandlerAdapter() {
                @Override
                public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
                    if (evt instanceof IdleStateEvent event) {
                        if (event.state() == IdleState.WRITER_IDLE) {
                            // 发送心跳包
                            logger.debug("produce 空闲，发送心跳至 broker...");
                            ctx.writeAndFlush(RemotingCommandProto.RemotingCommand.newBuilder().setType(MessageTypeEnum.IDLE_MESSAGE.getType()).build());
                        }
                    }
                }
            });
//            sc.pipeline().addLast(new ReconnectionHandler(BrokerConfig.NAME_SRV_ADDRESS, BrokerConfig.NAME_SRV_PORT));
            sc.pipeline().addLast(new SimpleChannelInboundHandler<RemotingCommand>() {
                @Override
                public void channelActive(ChannelHandlerContext ctx) {
                    logger.debug("连接成功");
                }

                @Override
                protected void channelRead0(ChannelHandlerContext channelHandlerContext, RemotingCommand remotingCommand) {
                    if (MessageTypeEnum.SUCCESS_MESSAGE.equals(MessageTypeEnum.fromType(remotingCommand.getType()))) {
                        RETRY_NUM_MAP.get(channelHandlerContext.channel().id().asLongText()).set(0);
                        logger.debug("当前 Broker 信息发送成功");
                    }
                    channelHandlerContext.fireChannelRead(remotingCommand);
                }

                @Override
                public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
                    logger.error("Broker 接收 NameSrv 數據 handler 異常");
                    logger.error(cause.getMessage());
                }
            });
        }
    });

    public static volatile boolean EXIT_FLAG = false;

    public static void start() {
        try {
            for (InetSocketAddress nameSrvAddress : BrokerConfig.NAME_SRV_ADDRESS) {
                Channel channel = client.connect(nameSrvAddress).sync().channel();
                channel.pipeline().addLast(new ReconnectionHandler(nameSrvAddress.getAddress().getHostAddress(), nameSrvAddress.getPort()));
                NAME_SRV_CHANNEL_MAP.put(channel.id().asLongText(), channel);
                RETRY_NUM_MAP.put(channel.id().asLongText(), new AtomicInteger());
                logger.info("连接 NameSrv {} 成功", nameSrvAddress.getAddress().getHostAddress() + ":" + nameSrvAddress.getPort());
            }
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }

        ThreadPoolHelper.DEFAULT_SCHEDULED_POOL.scheduleAtFixedRateByWatch(() -> {
            Iterator<Map.Entry<String, AtomicInteger>> iterator = RETRY_NUM_MAP.entrySet().iterator();
            while (iterator.hasNext()) {
                Map.Entry<String, AtomicInteger> entry = iterator.next();
                int num = entry.getValue().get();
                if (num >= 3) {
                    iterator.remove(); // 安全删除元素
                    NAME_SRV_CHANNEL_MAP.remove(entry.getKey()); // 同时删除关联的 Channel
                    logger.warn("NameSrv : {} 失败 {} 次, 失败3次后将关闭连接", entry.getKey(), num);
                }
            }

            if (NAME_SRV_CHANNEL_MAP.isEmpty() && EXIT_FLAG) {
                System.exit(0);
                logger.warn("NameSrv 以为空,关闭程序");
            }

            EXIT_FLAG = NAME_SRV_CHANNEL_MAP.isEmpty();
        }, 20, 20, TimeUnit.SECONDS);
    }


    public static ChannelFuture getChannel(String address, int port) {
        try {
            ChannelFuture channelFuture = client.connect(address, port);
            channelFuture.addListener((ChannelFutureListener) f -> {
                if (f.isSuccess()) {
                    logger.info("连接建立成功");

                } else {
                    logger.error("连接失败: {}", f.cause().getMessage());
                }
            });
            return channelFuture;
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    public static void sendBrokerAddress() {
        logger.info("准备发送 Broker 地址信息"); // 添加这行日志
        RETRY_NUM_MAP.values().forEach(AtomicInteger::getAndIncrement);

        logger.info("发送 Broker 信息到 NameServer: {}:{}", BrokerConfig.BROKER_ADDRESS, BrokerConfig.BROKER_PORT); // 添加详细信息


        NAME_SRV_CHANNEL_MAP.forEach((id, channel) -> NettyUtil.write(channel,
                RemotingCommand.builder()
                        .type(MessageTypeEnum.BROKER_INFO_MESSAGE.getType())
                        .body(JSONUtil.toJson(new BrokerInfo(BrokerConfig.BROKER_NAME, BrokerConfig.BROKER_ADDRESS, BrokerConfig.BROKER_PORT, QueueNumUtil.getMap())).getBytes(StandardCharsets.UTF_8))
                        .build()));
    }

    public static void addNameSrvChanel(Channel newChannel) {
        NAME_SRV_CHANNEL_MAP.put(newChannel.id().asLongText(), newChannel);
    }

    public static void close() {
        NAME_SRV_CHANNEL_MAP.forEach((id, channel) -> {
            try {
                channel.close().sync().addListener(f -> {
                    if (f.isSuccess()) {
                        logger.info("broker 关闭成功");
                    } else {
                        logger.error("broker 关闭异常");
                    }
                });
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        });
    }

    public static void closeChannel(String channelId) {
        Channel channel = NAME_SRV_CHANNEL_MAP.get(channelId);
        if (channel == null) {
            logger.error("错误关闭不存在的 NameSrv 连接");
            return;
        }

        try {
            channel.close().sync().addListener(f -> {
                if (f.isSuccess()) {
                    logger.info("broker 关闭成功");
                } else {
                    logger.error("broker 关闭异常");
                }
            });
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }

    }

    public static void removeChannel(String channelId) {
        NAME_SRV_CHANNEL_MAP.remove(channelId);
    }
}
